#include "io_handler.h"
#include "serial_surpport.h"
#ifdef DUMP_MEMORY_LEAK
#include <vld.h>
#endif
#include <boost/bind.hpp>
#include <boost/thread/locks.hpp>

#define DEFAULT_RECV_BUF 1460

void IoHandler::start()
{
	addr_ = socket_.remote_endpoint();
	socket_.set_option(tcp::no_delay(true));
	asyncReadHeader();
}

void IoHandler::asyncReadHeader()
{
	socket_.async_receive(recvBuf_.prepare(DEFAULT_RECV_BUF), 
		boost::bind(&IoHandler::tryReadHeader, 
		shared_from_this(), boost::asio::placeholders::bytes_transferred,
		boost::asio::placeholders::error));
}

void IoHandler::tryReadHeader(const size_t& bytes_transferred, 
	const boost::system::error_code& err)
{
	if(err)
	{
		DumpError(tryReadHeader, err, addr_);
		return;
	}
	if(closed_)
	{
		DumpCloseDetected(tryReadHeader, addr_);
		return;
	}
	recvBuf_.commit(bytes_transferred);
	istream is(&recvBuf_);

	while(recvBuf_.size() >= sizeof(Header))
	{	
		serial::read(is, curHeader_);
		if(static_cast<int>(recvBuf_.size()) < curHeader_.size_)
		{
			asyncReadPackage();
			return;
		}
		IPackage::Ptr package = IPackage::read(is, curHeader_);
		package->handle(shared_from_this());
		if(package.get() == NULL) { DumpUnKnownPak(curHeader_, addr_); _close(); return; }
	}
	asyncReadHeader();
}

void IoHandler::asyncReadPackage()
{
	boost::asio::async_read(socket_, recvBuf_, 
		boost::bind(&IoHandler::tryReadPackage, 
		shared_from_this(), boost::asio::placeholders::bytes_transferred,
		boost::asio::placeholders::error));
}

void IoHandler::tryReadPackage(const size_t& byte_transferred, const boost::system::error_code& err)
{
	if(err)
	{
		DumpError(tryReadPackage, err, addr_);
		return;
	}
	if(closed_)
	{
		DumpCloseDetected(tryReadPackage, addr_);
		return;
	}
	recvBuf_.commit(byte_transferred);
	istream is(&recvBuf_);
	while(static_cast<int>(recvBuf_.size()) >= curHeader_.size_)
	{	
		IPackage::Ptr package = IPackage::read(is, curHeader_);
		if(package.get() == NULL)
		{
			DumpUnKnownPak(curHeader_, addr_);
			_close();
			return;
		}
		if(recvBuf_.size() < sizeof(Header))
		{
			asyncReadHeader();
			return;
		}
		serial::read(is, curHeader_);
	}
	asyncReadPackage();
}

void IoHandler::send(IPackage::Ptr package)
{
	bool scheduleSend;
	{
		boost::unique_lock<boost::recursive_mutex> locker(sendingMutex_);
		scheduleSend = (sendBuf_.size() == 0);
		ostream os(&sendBuf_);
		IPackage::write(os, *package);
	}
	if(scheduleSend)
	{
		service_.post(boost::bind(&IoHandler::_send, shared_from_this()));
	}
}

void IoHandler::_send()
{
	socket_.async_send(sendBuf_.data(), boost::bind(&IoHandler::onWrite, 
		shared_from_this(), boost::asio::placeholders::bytes_transferred,
		boost::asio::placeholders::error));
}

void IoHandler::onWrite(const size_t& byte_transferred, const boost::system::error_code& err)
{
	if(err)
	{
		DumpError(onWrite, err, addr_);
		return;
	}
	if(closed_)
	{
		DumpCloseDetected(onWrite, addr_);
		return;
	}
	boost::unique_lock<boost::recursive_mutex> locker(sendingMutex_);
	sendBuf_.consume(byte_transferred);
	if(sendBuf_.size() > 0)
	{
		socket_.async_send(sendBuf_.data(), boost::bind(&IoHandler::onWrite, 
			shared_from_this(), boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error));
	}
}

void IoHandler::close()
{
	service_.post(boost::bind(&IoHandler::_close, 
		shared_from_this()));
}

void IoHandler::_close()
{
	DumpClose(addr_);
	closed_ = true;
	socket_.close();
}